/**
 * 对RabbitMQ的封装
 */
const amqp = require('amqplib');

class RabbitMQ {
  constructor() {
    this.hosts = [].concat(saas.config.rabbitmq.url);
    this.index = 0;
    this.length = this.hosts.length;
    this.open = amqp.connect(this.hosts[this.index]);
  }

  sendQueueMsg(queueName, msg, errCallBack) {
    let self = this;

    self.open
      .then(function (conn) {
        return conn.createChannel();
      })
      .then(function (channel) {
        return channel.assertQueue(queueName).then(function (ok) {
          return channel.sendToQueue(queueName, new Buffer(msg), {
            persistent: true
          });
        })
          .then(function (data) {
            if (data) {
              errCallBack && errCallBack("success");
              channel.close();
            }
          })
          .catch(function () {
            setTimeout(() => {
              if (channel) {
                channel.close();
              }
            }, 500)
          });
      })
      .catch(function () {
        let num = self.index++;

        if (num <= self.length - 1) {
          self.open = amqp.connect(self.hosts[num]);
        } else {
          self.index = 0;
        }
      });
  }

  receiveQueueMsg(queueName, receiveCallBack, errCallBack) {
    let self = this;

    self.open
      .then(function (conn) {
        return conn.createChannel();
      })
      .then(function (channel) {
        return channel.assertQueue(queueName)
          .then(function (ok) {
            return channel.consume(queueName, function (msg) {
              if (msg !== null) {
                let data = msg.content.toString();
                channel.ack(msg);
                receiveCallBack && receiveCallBack(data);
              }
            })
              .finally(function () {
                setTimeout(() => {
                  if (channel) {
                    channel.close();
                  }
                }, 500)
              });
          })
      })
      .catch(function () {
        let num = self.index++;
        if (num <= self.length - 1) {
          self.open = amqp.connect(self.hosts[num]);
        } else {
          self.index = 0;
          self.open = amqp.connect(self.hosts[0]);
        }
      });
  }

  async sendToDLX(exchangeName, queueName, exchangeDLXName, routingKeyDLX, msg, expiration) {
    let self = this;

    const conn = await self.open;
    const ch = await conn.createChannel();
    await ch.assertExchange(exchangeName, 'direct', {durable: true});
    const queueResult = await ch.assertQueue(queueName, {
      exclusive: false,
      deadLetterExchange: exchangeDLXName,
      deadLetterRoutingKey: routingKeyDLX,
    });
    await ch.bindQueue(queueResult.queue, exchangeName);
    const res = await ch.sendToQueue(queueResult.queue, new Buffer(msg), {
      expiration: expiration
    });
    console.log(res);
    ch.close();
  }

  async consumeDLX(queueName, exchangeDLXName, routingKeyDLX, cb) {
    let self = this;
    const conn = await self.open;
    const ch = await conn.createChannel();
    await ch.assertExchange(exchangeDLXName, 'direct', {durable: true});
    const queueResult = await ch.assertQueue(queueName, {
      exclusive: false,
    });
    await ch.bindQueue(queueResult.queue, exchangeDLXName, routingKeyDLX);
    await ch.consume(queueResult.queue, msg => {
      cb && cb(msg.content.toString());
    }, {noAck: true});
  }

  async sendDelayMsg(exchangeName, queueName, msg, expiration) {
    let self = this;

    const conn = await self.open;
    const ch = await conn.createChannel();
    await ch.assertExchange(exchangeName, 'x-delayed-message', {durable: true, arguments: {'x-delayed-type': 'direct'}});
    const queueResult = await ch.assertQueue(queueName);
    await ch.bindQueue(queueResult.queue, exchangeName);
    const res = await ch.publish(exchangeName, '', new Buffer(msg), {
      headers: {'x-delay': expiration}
    });
    console.log(res);
    ch.close();
  }

  async consume(queueName, cb) {
    let self = this;
    const conn = await self.open;
    const ch = await conn.createChannel();
    const queueResult = await ch.assertQueue(queueName);
    await ch.consume(queueResult.queue, msg => {
      cb && cb(msg.content.toString());
    }, {noAck: true});
  }

}

module.exports = RabbitMQ;